package com.shujia.opt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo4Coalesce1 {
  def main(args: Array[String]): Unit = {
    //repartition:coalesce(numPartitions，true) 增多分区使用这个
    //coalesce(numPartitions，false) 减少分区 没有shuffle只是合并 partition
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("重分区")
    val sc = new SparkContext(conf)

    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    println(s"lineRDD的分区数：${lineRDD.getNumPartitions}")

    /**
     * 使用repartition
     */
    //增大分区数，使用repartition，返回一个新的rdd,会产生shuffle
    //    val resRDD1: RDD[String] = lineRDD.repartition(10)
    //    println(s"resRDD1的分区数：${resRDD1.getNumPartitions}")
    //    resRDD1.foreach(println)
    //减少分区数，使用repartition，返回一个新的rdd,会产生shuffle
    //    val resRDD2: RDD[String] = resRDD1.repartition(1)
    //    println(s"resRDD2的分区数：${resRDD2.getNumPartitions}")
    //    resRDD2.foreach(println)


    /**
     * coalesce
     *
     * 1、默认增大分区是不会产生shuffle的
     * 2、合并分区直接给分区数，不会产生shuffle
     */

    val resRDD1: RDD[String] = lineRDD.coalesce(10, shuffle = true)
    println(s"resRDD1的分区数：${resRDD1.getNumPartitions}")
//    resRDD1.foreach(println)

    val resRDD2: RDD[String] = resRDD1.coalesce(1,shuffle = true)
    println(s"resRDD2的分区数：${resRDD2.getNumPartitions}")
    resRDD2.foreach(println)


    while (true) {

    }


  }
}
